package RabbitMQ

import (
	"fmt"
	"github.com/streadway/amqp"
	"log"
)

//url 格式  amqp://账号：密码@rabiitmq服务器地址:端口号/vhost
const MRURL = "amqp://imoocuser:imoocpass@127.0.0.1:5672/imooc"

type RabbitMQ struct {
	conn    *amqp.Connection
	channel *amqp.Channel
	//队列名
	QueueName string
	//交换机
	Exchange string
	//key
	Key string
	//链接信息
	Mqurl string
}

//创建RabbitMQ结构体实例(基础实例)
func NewRabbitMQ(queueName string, exchange string, key string) *RabbitMQ {
	return &RabbitMQ{
		QueueName: queueName,
		Exchange:  exchange,
		Key:       key,
		Mqurl:     MRURL,
	}
}

//断开channel和connection的链接
func (r RabbitMQ) Destory() {
	r.channel.Close()
	r.conn.Close()
}

//错误处理函数（创建结构体后，一定要处理错误）
func (r *RabbitMQ) failOnErr(err error, message string) {
	if err != nil {
		log.Fatalf("%s:%s", message, err)
		panic(fmt.Sprintf("%s:%s", message, err))
	}
}

/**
rabbitmq简单模式
*/
//步骤一：创建rabbitmq的简单模式实例
func NewRabbitMQSimple(queueName string) *RabbitMQ {
	rabbitmq := NewRabbitMQ(queueName, "", "")
	var err error
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnErr(err, "创建链接失败！")
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "创建channel失败")
	return rabbitmq
}

//步骤二：简单模式下的生产
func (r *RabbitMQ) PublishSimple(message string) {
	//1.申请队列，如果队列不存在就创建，如果存在就跳过创建
	//保证 消息能够入队
	_, err := r.channel.QueueDeclare(
		r.QueueName, //队列名称
		false,       //是否持久化
		false,       //自动删除（最后一个消费者断开链接后删除队列）
		false,       //是否排他
		false,       //是否柱塞
		nil,         //额外属性
	)
	if err != nil {
		fmt.Println(err)
	}
	//2.发送消息到队列中
	r.channel.Publish(
		r.Exchange,  //交换机
		r.QueueName, //队列名称
		false,       //如果为true回根据exchange和routekey的规则，如果无法找到符合规则的队列，就会把消息返回生产者
		false,       //如果为true，当exchange发现队列没有绑定消费者，就会把消息返回给生产者
		//发送的消息
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message), //消息内容
		})
}

//步骤三：简单模式下的消费
func (r *RabbitMQ) ConsumeSimple() {
	//1.申请队列，如果队列不存在就创建，如果存在就跳过创建
	//保证 消息能够入队
	_, err := r.channel.QueueDeclare(
		r.QueueName, //队列名称
		false,       //是否持久化
		false,       //自动删除（最后一个消费者断开链接后删除队列）
		false,       //是否排他
		false,       //是否柱塞
		nil,         //额外属性
	)
	if err != nil {
		fmt.Println(err)
	}
	//2.从队列中获取消息
	msgs, err := r.channel.Consume(
		r.QueueName, //队列名称
		"",          //用来区分多个消费者
		true,        //是否自动应答（默认为true）
		false,       //是否具有排他性
		false,       //如果为true表示，不能将同一个connection中发送的消息，传递这个connection中的消费者
		false,       //是否柱塞
		nil,         //其他消息
	)
	if err != nil {
		fmt.Println(err)
	}
	//3。消费队列消息
	forever := make(chan bool)
	go func() {
		//消费队列的逻辑处理
		//入库及其他处理
		for d := range msgs {
			log.Printf("消费了一个消息：%s", d.Body)
		}
	}()
	log.Printf("等待消息，退出按control + C")
	<-forever //柱塞，等待队列有消息
}

/**
rabbitmq消息订阅模式（一个消息可以被很多个消费者消费）
*/
//步骤一：创建rabbitmq的订阅模式实例
func NewRabbitMQPublish(exchangeName string) *RabbitMQ {
	rabbitmq := NewRabbitMQ("", exchangeName, "")
	var err error
	//获取connection
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnErr(err, "创建链接失败!")
	//获取Chanel
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "创建channel失败")
	return rabbitmq
}

//步骤二：订阅模式下的生产
func (r *RabbitMQ) PublishPub(message string) {
	//1.尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		"fanout", //广播类型
		true,     //是否持久化
		false,    //是否自动删除
		false,    //YES表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定，一般设置为false
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare an exchange")
	//2.发送消息
	err = r.channel.Publish(
		r.Exchange,
		"",
		false, //如果为true回根据exchange和routekey的规则，如果无法找到符合规则的队列，就会把消息返回生产者
		false, //如果为true，当exchange发现队列没有绑定消费者，就会把消息返回给生产者
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
}

//步骤三：订阅模式下的消费
func (r *RabbitMQ) ReceivePub() {
	//1.尝试创建交换机/Users/yuzhonghua/go/src/imooc-rabbitmq
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		"fanout",
		true,  //是否持久化
		false, //是否自动删除
		false, //YES表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定，一般设置为false
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare an exchange")
	//2.尝试创建一个队列，这里注意队列名称不要写
	declare, err := r.channel.QueueDeclare(
		"", //队列名称为空，表示随机创建队列
		false,
		false,
		true, //排他性，设置为true
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare a queue")
	//3.绑定队列到exchange中/Users/yuzhonghua/go/src/imooc-rabbitmq
	err = r.channel.QueueBind(
		declare.Name, //队列名称
		"",           //在pub/sub模式下，这里的key要为空
		r.Exchange,   //交换机名称
		false,
		nil,
	)
	//消费消息
	consume, err := r.channel.Consume(
		declare.Name,
		"",
		true,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		fmt.Println(err.Error())
	}
	forever := make(chan bool)
	go func() {
		for d := range consume {
			log.Printf("Received a message :%s", d.Body)
		}
	}()
	log.Printf("等待消息，退出按control + C")
	<-forever //等待消息柱塞
}

/**
rabbitmq的路由模式
*/
//第一步：创建实例
func NewRabbitMQRouting(exchangeName string, routingtkey string) *RabbitMQ {
	rabbitmq := NewRabbitMQ("", exchangeName, routingtkey)
	var err error
	//获取connection
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnErr(err, "创建链接失败!")
	//获取Chanel
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "创建channel失败")
	return rabbitmq
}

//第二步：routing模式下的消息发送
func (r *RabbitMQ) PublishRouting(message string) {
	//1.尝试创建交换机/Users/yuzhonghua/go/src/imooc-rabbitmq
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		"direct", //与订阅模式fanout不同，这里为direct
		true,     //是否持久化
		false,    //是否自动删除
		false,    //YES表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定，一般设置为false
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare an exchange")
	//2。发送消息
	err = r.channel.Publish(
		r.Exchange,
		r.Key,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
}

//第三步：routing模式下的消息消费
func (r *RabbitMQ) ReceiveRouting() {
	//1.尝试创建交换机/Users/yuzhonghua/go/src/imooc-rabbitmq
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		"direct",
		true,  //是否持久化
		false, //是否自动删除
		false, //YES表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定，一般设置为false
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare an exchange")
	//申明队列
	//2.尝试创建一个队列，这里注意队列名称不要写
	declare, err := r.channel.QueueDeclare(
		"", //队列名称为空，表示随机创建队列
		false,
		false,
		true, //排他性，设置为true
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare a queue")
	//3.绑定队列到exchange中/Users/yuzhonghua/go/src/imooc-rabbitmq
	err = r.channel.QueueBind(
		declare.Name, //队列名称
		r.Key,        //在pub/sub模式下，这里的key要为空
		r.Exchange,   //交换机名称
		false,
		nil,
	)
	//消费消息
	consume, err := r.channel.Consume(
		declare.Name,
		"",
		true,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		fmt.Println(err.Error())
	}
	forever := make(chan bool)
	go func() {
		for d := range consume {
			log.Printf("Received a message :%s", d.Body)
		}
	}()
	log.Printf("等待消息，退出按control + C")
	<-forever //等待消息柱塞
}
